#ifndef CALLBACK_CLIENT_H_
#define CALLBACK_CLIENT_H_
#include <string>
#include <exception>
#include <boost/thread.hpp>
#include <boost/shared_ptr.hpp>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TDebugProtocol.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/THttpClient.h>
#include <thrift/transport/THttpServer.h>
#include <thrift/transport/TTransport.h>
#include <thrift/transport/TBufferTransports.h>
#include <thrift/TProcessor.h>
#include <glog/logging.h>
#include "gen-cpp/PushDemo.h"
namespace at = apache::thrift;
namespace atp = apache::thrift::protocol;
namespace att = apache::thrift::transport;
namespace client {
  
  struct Notify {
    volatile bool connected_;
    boost::condition_variable disconnected_cond_;
    boost::condition_variable connected_cond_;
    boost::mutex mutex_;
  };
  
class Callable {
private:
  enum ErrorType { OK, GENERIC_ERROR, TRANS_EXCEPTION};
 public:
Callable(boost::shared_ptr<at::TProcessor> processor, boost::shared_ptr<atp::TProtocol> in_out, Notify* notify, boost::shared_ptr<att::TSocket> sock) : processor_(processor), in_out_(in_out), notify_(notify)/*, client_(new push::PushDemoClient(in_out_))*/ {
    boost::shared_ptr<att::TTransport> trans(new att::TFramedTransport(sock));
    const std::string http_header_host("www.cfood.com");
    const std::string path("/");
    boost::shared_ptr<att::TTransport> http(new att::THttpClient(trans, http_header_host,
                                                                 path));
    boost::shared_ptr<atp::TBinaryProtocol> prot(new atp::TBinaryProtocol(http));
    client_.reset(new push::PushDemoClient(prot));
  }
  void operator()() {
    for (;;) {
      {
        boost::unique_lock<boost::mutex> lock(notify_->mutex_);
        while (!notify_->connected_) {
          LOG(INFO) << "wait for reconnect ... ";       
          notify_->connected_cond_.wait(lock);
        }
      }
      // connected
      for (;;) {
        ErrorType rc = SendEmptyReq();
        if (Error(rc)) {
          break;
        }
        rc = HandleResp();
        if (Error(rc)) {
          break;
        }
      } // end inner for loop
    } // end outer for loop
  }
  bool Error(ErrorType rc) {
    if (rc >= GENERIC_ERROR) {
      if (rc >= TRANS_EXCEPTION) {
        boost::unique_lock<boost::mutex> lock(notify_->mutex_);
        notify_->connected_ = false;
        notify_->disconnected_cond_.notify_one();
      }
      return true;
    }
    return false;
  }
  ErrorType SendEmptyReq() {
    // send empty request
    try {
      push::EmptyRequest req;
      client_->CSSendEmptyRequest(req);
    } catch (const att::TTransportException& tte) {
      LOG(ERROR) << "client send empty request transport exception: " << tte.what();
      return TRANS_EXCEPTION;
    }  catch (const std::exception& e) {
      LOG(ERROR) << "client send empty request failed: " << e.what();
      return GENERIC_ERROR;
    }
    return OK;
  }
  ErrorType HandleResp() {
    // handle response
    try {
      if (!processor_->process(in_out_, in_out_, NULL) ||
          !in_out_->getTransport()->peek()) {
        LOG(ERROR) << "process error or transport peek error";
        return GENERIC_ERROR;
      }
    } catch (const att::TTransportException& tte) {
      LOG(ERROR) << "handle response transport exception: " << tte.what();
      return TRANS_EXCEPTION;
    }  catch (const std::exception& e) {
      LOG(ERROR) << "handle response failed: " << e.what();
      return GENERIC_ERROR;
    }
    return OK;
  }
 private:
  boost::shared_ptr<at::TProcessor> processor_;
  boost::shared_ptr<atp::TProtocol> in_out_;
  Notify* notify_;
  boost::shared_ptr<push::PushDemoClient> client_;
};

class ReconnectCallable {
public:
  ReconnectCallable(boost::shared_ptr<att::TTransport> trans, Notify* notify)
      : trans_(trans), notify_(notify) {
    
  }
  void operator()() {
    for (;;) {
      {
        boost::unique_lock<boost::mutex> lock(notify_->mutex_);
        if (notify_->connected_) {
          notify_->disconnected_cond_.wait(lock);
        }
      }
      LOG(INFO) << "try reconnect ... ";
      try {
        trans_->close();
        trans_->open();
      } catch (const std::exception& e) {
        LOG(ERROR) << "open transport failed: " << e.what();
        sleep(5);
        continue;
      }
      LOG(INFO) << "open transport success";
      {
        boost::unique_lock<boost::mutex> lock(notify_->mutex_);
        notify_->connected_ = true;
        notify_->connected_cond_.notify_one();
      }
    }
  }
private:
  boost::shared_ptr<att::TTransport> trans_;
  Notify* notify_;
};
template <typename HandlerType, typename ProcessorType>
  class CallbackClient {
 public:
  CallbackClient(const std::string& host, int port, const std::string& path) {
    sock_.reset(new att::TSocket(host, port));
    trans_.reset(new att::THttpClient(sock_, host, path));
    prot_.reset(new atp::TBinaryProtocol(trans_));
    notify_.connected_ = false;
  }
  void Start() {
    // 
    boost::shared_ptr<HandlerType> handler(new HandlerType());
    boost::shared_ptr<at::TProcessor> processor(new ProcessorType(handler));
    thread_.reset(new boost::thread(Callable(processor, prot_, &notify_, sock_)));
    reconnect_thread_.reset(new boost::thread(ReconnectCallable(trans_, &notify_)));
    thread_->join();
    reconnect_thread_->join();
  }
 private:
  boost::shared_ptr<att::TSocket> sock_;
  boost::shared_ptr<att::TTransport> trans_;
  boost::shared_ptr<atp::TProtocol> prot_;
  Notify notify_;
  boost::shared_ptr<boost::thread> thread_;
  boost::shared_ptr<boost::thread> reconnect_thread_;
};
}
#endif
